package com.gitee.Jmysy.binlog4j.core;

import com.alibaba.fastjson.parser.ParserConfig;
import com.alibaba.fastjson.util.TypeUtils;
import com.gitee.Jmysy.binlog4j.core.meta.ColumnMetadata;
import com.gitee.Jmysy.binlog4j.core.utils.ClassUtils;
import com.gitee.Jmysy.binlog4j.core.utils.JDBCUtils;
import lombok.Data;
import lombok.SneakyThrows;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Binlog 处理器包装类
 *
 * @author 就眠儀式
 */
@Data
public class BinlogEventHandlerInvoker<T> {

    private static final ParserConfig SNAKE_CASE;

    static {
        SNAKE_CASE = new ParserConfig();
        //        SNAKE_CASE.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
    }

    private Map<String, List<ColumnMetadata>> columnMetadataMap = new HashMap<>();
    private IBinlogEventHandler eventHandler;
    private BinlogClientConfig clientConfig;
    private Class<T> genericClass;

    public void invokeInsert(String databaseName, String tableName, List<Serializable[]> data) {
        if (eventHandler.isHandle(databaseName, tableName)) {
            List<ColumnMetadata> columns = getColumns(databaseName, tableName);
            List<String> primaryKeyNames = columns.stream().filter(ColumnMetadata::getIsPrimaryKey)
                    .map(ColumnMetadata::getColumnName).collect(Collectors.toList());
            BinlogEvent binlogEvent = createBinlogEvent(databaseName, tableName);
            binlogEvent.setPrimaryKeys(primaryKeyNames);
            data.forEach(row -> {
                binlogEvent.setData(toEntity(columns, row));
                eventHandler.onInsert(binlogEvent);
            });
        }
    }

    public void invokeUpdate(String databaseName, String tableName, List<Map.Entry<Serializable[], Serializable[]>> data) {
        if (eventHandler.isHandle(databaseName, tableName)) {
            List<ColumnMetadata> columns = getColumns(databaseName, tableName);
            List<String> primaryKeyNames = columns.stream().filter(ColumnMetadata::getIsPrimaryKey)
                    .map(ColumnMetadata::getColumnName).collect(Collectors.toList());
            BinlogEvent binlogEvent = createBinlogEvent(databaseName, tableName);
            binlogEvent.setPrimaryKeys(primaryKeyNames);
            data.forEach(row -> {
                binlogEvent.setData(toEntity(columns, row.getValue()));
                binlogEvent.setOriginalData(toEntity(columns, row.getKey()));
                eventHandler.onUpdate(binlogEvent);
            });
        }
    }

    public void invokeDelete(String databaseName, String tableName, List<Serializable[]> data) {
        if (eventHandler.isHandle(databaseName, tableName)) {
            List<ColumnMetadata> columns = getColumns(databaseName, tableName);
            List<String> primaryKeyNames = columns.stream().filter(ColumnMetadata::getIsPrimaryKey)
                    .map(ColumnMetadata::getColumnName).collect(Collectors.toList());
            BinlogEvent binlogEvent = createBinlogEvent(databaseName, tableName);
            binlogEvent.setPrimaryKeys(primaryKeyNames);
            data.forEach(row -> {
                binlogEvent.setData(toEntity(columns, row));
                eventHandler.onDelete(binlogEvent);
            });
        }
    }

    private BinlogEvent createBinlogEvent(String databaseName, String tableName) {
        BinlogEvent binlogEvent = new BinlogEvent<>();
        binlogEvent.setDatabase(databaseName);
        binlogEvent.setTable(tableName);
        binlogEvent.setTimestamp(System.currentTimeMillis());
        return binlogEvent;
    }

    public List<ColumnMetadata> getColumns(String databaseName, String tableName) {
        String tableSchema = String.format("%s.%s", databaseName, tableName);
        List<ColumnMetadata> columns = columnMetadataMap.get(tableSchema);
        if (columns == null || clientConfig.isStrict()) {
            columns = JDBCUtils.getColumns(clientConfig, databaseName, tableName);
            columnMetadataMap.put(tableSchema, columns);
        }
        return columns;
    }

    @SneakyThrows
    public T toEntity(List<ColumnMetadata> columns, Serializable[] data) {
        Map<String, Object> obj = new HashMap<>(columns.size());
        for (int i = 0; i < data.length; i++) {
            ColumnMetadata column = columns.get(i);
            Serializable fieldValue = data[i];
            if (fieldValue instanceof java.util.Date) {
                if (fieldValue != null) {
                    data[i] = new Date(((Date) fieldValue).getTime() + clientConfig.getTimeOffset());
                }
            } else if (fieldValue instanceof byte[]) {
                if (fieldValue != null) {
                    if (genericClass != null) {
                        Field field = ClassUtils.getDeclaredField(genericClass, column.getColumnName());
                        if (field != null) {
                            if (field.getType() == String.class) {
                                data[i] = new String((byte[]) fieldValue, StandardCharsets.UTF_8);
                            }
                        }
                    } else {
                        // 非泛型
                        String dataType = column.getDataType();
                        if ("text".equals(dataType)) {
                            String value = new String((byte[]) fieldValue, StandardCharsets.UTF_8);
                            data[i] = value;
                        }
                    }
                }
            } else if (fieldValue instanceof BitSet) {
                if (fieldValue != null) {
                    if (genericClass != null) {
                        Field field = ClassUtils.getDeclaredField(genericClass, column.getColumnName());
                        if (field != null) {
                            if (field.getType() == Boolean.class || field.getType() == boolean.class) {
                                data[i] = !((BitSet) fieldValue).isEmpty();
                            }
                        }
                    }
                }
            }
            obj.put(column.getColumnName(), data[i]);
        }
        if (genericClass != null) {
            return TypeUtils.cast(obj, genericClass, SNAKE_CASE);
        }
        return (T) obj;
    }

    public void setEventHandler(IBinlogEventHandler eventHandler) {
        this.eventHandler = eventHandler;
        this.genericClass = ClassUtils.getGenericType(eventHandler.getClass());
    }
}
